package com.tlb.springboot.starter.rocketmq.impl;

import com.aliyun.openservices.shade.com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.SendResult;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.SendStatus;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.Message;
import com.aliyun.openservices.shade.com.alibaba.rocketmq.remoting.RPCHook;
import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
import com.tlb.springboot.starter.rocketmq.message.DefaultMessage;
import com.tlb.springboot.starter.trace.Constants;
import com.tlb.springboot.starter.trace.context.TraceContext;

import lombok.extern.slf4j.Slf4j;

import javax.annotation.PreDestroy;
import java.util.HashMap;
import java.util.Map;

/**
 * @author yuaneg 2018/8/30
 */

@Slf4j
public class RocketMQProducer extends DefaultMQProducer {

    private static final int DEFAULT_DELAY_TIME_LEVEL = 16;
    private boolean isAliyun = false;

    public RocketMQProducer() {
        super();
    }

    public RocketMQProducer(String namespace, String producerGroup, RPCHook rpcHook) {
        super(namespace, producerGroup, rpcHook);
        isAliyun = true;
    }

    public RocketMQProducer(String producerGroup) {
        super(producerGroup);
    }

    public RocketMQProducer(RPCHook rpcHook) {
        super(rpcHook);
    }

    /**
     * @param message 要发送的消息
     */
    public <T extends Message> boolean sendMsg(T message) {
        SendResult sendResult = null;
        try {
            //测试等自建rocketMQ环境, 默认延时30分钟
            if (isAliyun == false && message instanceof DefaultMessage
                && ((DefaultMessage) message).getStartDeliverTime() != 0) {
                message.setDelayTimeLevel(DEFAULT_DELAY_TIME_LEVEL);
            }
            setTraceId(message);
            sendResult = super.send(message);
        } catch (Exception e) {
            Map<String, Object> json = new HashMap<>();
            json.put("rocketMQ", "发送消息异常");
            json.put("errorMsg", e.getMessage());
            log.error(JSONObject.toJSONString(json), e);
            return false;
        }
        return sendResult != null && sendResult.getSendStatus() == SendStatus.SEND_OK;
    }

    public boolean sendMsg(String topic, Object body) {
        DefaultMessage message = new DefaultMessage(topic, body);
        return sendMsg(message);
    }

    /**
     * 发送延时消息
     *
     * @param message    message
     * @param delayLevel default 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * @return true/false
     */
    public <T extends Message> boolean sendDelay(T message, int delayLevel) {
        SendResult sendResult = null;
        try {
            message.setDelayTimeLevel(delayLevel);
            setTraceId(message);
            sendResult = super.send(message);
        } catch (Exception e) {
            Map<String, Object> json = new HashMap<>();
            json.put("rocketMQ", "发送delay消息异常");
            json.put("errorMsg", e.getMessage());
            log.error(JSONObject.toJSONString(json), e);
            return false;
        }
        return sendResult != null && sendResult.getSendStatus() == SendStatus.SEND_OK;
    }

    public boolean sendDelay(String topic, Object body, int delayLevel) {
        DefaultMessage message = new DefaultMessage(topic, body);
        return sendDelay(message, delayLevel);
    }

    private <T extends Message> void setTraceId(T message) {
        String traceId = TraceContext.getTraceId();
        if (StringUtils.isNoneBlank(traceId)) {
            message.putUserProperty(Constants.TRACE_ID, traceId);
        }
    }

    @PreDestroy
    public void destroy() {
        super.shutdown();
    }
}
